时会|都会_Netty框架的基本使用

篇首语：本文由编程笔记#小编为大家整理，主要介绍了Netty框架的基本使用相关的知识，希望对你有一定的参考价值。

为什么使用Netty

前面了解过的NIO模型&＃xff0c;它有可靠性高、吞吐量高的优点&＃xff0c;但也存在编程复杂的问题&＃xff0c;我们要掌握大量的API&＃xff0c;如&＃xff1a;各种Channel、Buffer、Selector&＃xff0c;还要编程处理特殊情况&＃xff0c;如&＃xff1a;异常处理、断线重连、心跳包等。

Netty对JDK自带的NIO包进行了大量的封装&＃xff0c;较好的解决了上述问题&＃xff0c;让我们既获得NIO模型性能高、资源消耗小、可靠性高等优点&＃xff0c;又降低了编程的难度&＃xff0c;所以目前进行网络编程时常用Netty框架。

Netty的应用场景

1. 网络游戏服务器

   网络游戏服务器领域也大量使用了Java语言&＃xff0c;Netty作为高性能的网络通信框架&＃xff0c;得到了广泛的应用。

2. 互联网应用

   分布式系统中各个节点的通信都可以使用Netty实现&＃xff0c;如著名的RPC框架 Dubbo以及消息中间件 RocketMQ内部都采用了Netty。

3. 大数据

   Hadoop体系中的Avro数据序列化系统&＃xff0c;采用了Netty实现的RPC服务器

maven依赖


io.netty
netty-all
4.1.35.Final



服务器

public class NettyServer
public static void main(String[] args) throws Exception
//处理连接请求的线程组
EventLoopGroup bossGroup &＃61; new NioEventLoopGroup(2);
//处理客户端业务的线程组
EventLoopGroup workerGroup &＃61; new NioEventLoopGroup(10);
try
//创建服务器端的启动对象
ServerBootstrap bootstrap &＃61; new ServerBootstrap();
//设置两个线程组
bootstrap.group(bossGroup, workerGroup)
//使用NioserverSocketChannel作为服务器的通道实现
.channel(NioServerSocketChannel.class)
// 初始化服务器连接队列大小&＃xff0c;服务端将不能处理的客户端连接请求放在队列中等待处理
.option(ChannelOption.SO_BACKLOG, 1024)
// 创建通道初始化对象&＃xff0c;设置初始化参数
.childHandler(new ChannelInitializer()
&＃64;Override
protected void initChannel(SocketChannel ch) throws Exception
//对workerGroup的SocketChannel设置处理器
ch.pipeline().addLast(new NettyServerHandler());

);
System.out.println("Server start");
//启动服务器&＃xff0c;bind是异步操作&＃xff0c;sync方法是等待异步操作执行完毕
ChannelFuture cf &＃61; bootstrap.bind(9000).sync();
// 对通道关闭进行监听&＃xff0c;closeFuture是异步操作&＃xff0c;通过sync方法同步等待通道关闭处理完毕
cf.channel().closeFuture().sync();
finally
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();




服务器的处理器

/**
* 自定义Handler
*/
public class NettyServerHandler extends ChannelInboundHandlerAdapter
/**
* 读取客户端发送的数据
*/
&＃64;Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
System.out.println("服务器读取线程 " &＃43; Thread.currentThread().getName());
//将 msg 转成一个 ByteBuf&＃xff0c;类似NIO 的 ByteBuffer
ByteBuf buf &＃61; (ByteBuf) msg;
System.out.println("客户端发送消息是:" &＃43; buf.toString(CharsetUtil.UTF_8));

/**
* 数据读取完毕处理方法
*/
&＃64;Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception
ByteBuf buf &＃61; Unpooled.copiedBuffer("Hello Client".getBytes(CharsetUtil.UTF_8));
ctx.writeAndFlush(buf);

/**
* 处理异常
*/
&＃64;Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception
ctx.close();



客户端

public class NettyClient
public static void main(String[] args) throws Exception
//客户端线程组
EventLoopGroup group &＃61; new NioEventLoopGroup();
try
//创建客户端启动对象
Bootstrap bootstrap &＃61; new Bootstrap();
//设置线程组
bootstrap.group(group)
// 使用NioSocketChannel作为客户端的通道实现
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer()
&＃64;Override
protected void initChannel(SocketChannel ch) throws Exception
//加入处理器
ch.pipeline().addLast(new NettyClientHandler());

);
System.out.println("Client start");
//启动客户端去连接服务器端
ChannelFuture cf &＃61; bootstrap.connect("127.0.0.1", 9000).sync();
//对通道关闭进行监听
cf.channel().closeFuture().sync();
finally
group.shutdownGracefully();




客户端处理器

public class NettyClientHandler extends ChannelInboundHandlerAdapter
/**
* 客户端连接服务器完成
*/
&＃64;Override
public void channelActive(ChannelHandlerContext ctx) throws Exception
ByteBuf buf &＃61; Unpooled.copiedBuffer("Hello Server".getBytes(CharsetUtil.UTF_8));
ctx.writeAndFlush(buf);

/**
* 当通道有读取事件时会触发&＃xff0c;即服务端发送数据给客户端
*/
&＃64;Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
ByteBuf buf &＃61; (ByteBuf) msg;
System.out.println("收到服务端的消息:" &＃43; buf.toString(CharsetUtil.UTF_8));
System.out.println("服务端的地址&＃xff1a; " &＃43; ctx.channel().remoteAddress());

&＃64;Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception
cause.printStackTrace();
ctx.close();


Netty工作模型

工作模型介绍

1. Netty 抽象出两组线程池BossGroup和WorkerGroup&＃xff0c;BossGroup专门负责接收客户端的连接,
   WorkerGroup专门负责网络的读写
2. BossGroup和WorkerGroup类型都是NioEventLoopGroup
3. NioEventLoopGroup 相当于一个事件循环线程组, 这个组中含有多个事件循环线程&＃xff0c;每一个事件
   循环线程是NioEventLoop
4. 每个NioEventLoop都有一个selector , 用于监听注册在其上的socketChannel的网络通讯
5. 每个Boss NioEventLoop线程内部循环执行的步骤有 3 步
   
   1. 处理accept事件 , 与client 建立连接 , 生成 NioSocketChannel
   2. 将NioSocketChannel注册到某个worker NIOEventLoop上的selector
   3. 处理任务队列的任务 &＃xff0c; 即runAllTasks
6. 每个worker NIOEventLoop线程循环执行的步骤
   
   1. 轮询注册到自己selector上的所有NioSocketChannel 的read, write事件
   2. 处理 I/O 事件&＃xff0c; 即read , write 事件&＃xff0c; 在对应NioSocketChannel 处理业务
   3. runAllTasks处理任务队列TaskQueue的任务 &＃xff0c;一些耗时的业务处理一般可以放入TaskQueue中慢慢处理&＃xff0c;这样不影响数据在 pipeline 中的流动处理
7. 每个worker NIOEventLoop处理NioSocketChannel业务时&＃xff0c;会使用 pipeline (管道)&＃xff0c;管道中维护
   了很多 handler 处理器用来处理 channel 中的数据

• Bootstrap、ServerBootstrap
  Bootstrap 意思是引导&＃xff0c;一个 Netty 应用通常由一个 Bootstrap 开始&＃xff0c;主要作用是配置整个 Netty 程序&＃xff0c;串联各个组件&＃xff0c;Netty 中 Bootstrap 类是客户端程序的启动引导类&＃xff0c;ServerBootstrap 是服务端启动引导类。

• Future、ChannelFuture
  在 Netty 中所有的 IO 操作都是异步的&＃xff0c;不能立刻得知消息是否被正确处理。
  但是可以过一会等它执行完成或者直接注册一个监听&＃xff0c;具体的实现就是通过 Future 和ChannelFutures&＃xff0c;他们可以注册一个监听&＃xff0c;当操作执行成功或失败时监听会自动触发注册的监听事件。

• Channel
  Netty 网络通信的组件&＃xff0c;能够用于执行网络 I/O 操作。

  Channel 为用户提供&＃xff1a;
  1&＃xff09;当前网络连接的通道的状态&＃xff08;例如是否打开&＃xff1f;是否已连接&＃xff1f;&＃xff09;
  2&＃xff09;网络连接的配置参数 &＃xff08;例如接收缓冲区大小&＃xff09;
  3&＃xff09;提供异步的网络 I/O 操作(如建立连接&＃xff0c;读写&＃xff0c;绑定端口)&＃xff0c;异步调用意味着任何 I/O 调用都将立即返回&＃xff0c;且不保证在调用结束时所请求的 I/O 操作已完成。
  4&＃xff09;调用立即返回一个 ChannelFuture 实例&＃xff0c;通过注册监听器到 ChannelFuture 上&＃xff0c;可以 I/O 操作成功、失败或取消时回调通知调用方。
  5&＃xff09;支持关联 I/O 操作与对应的处理程序。
  不同协议、不同的阻塞类型的连接都有不同的 Channel 类型与之对应。
  下面是一些常用的 Channel 类型&＃xff1a;
  1 NioSocketChannel&＃xff0c;异步的客户端 TCP Socket 连接。
  2 NioServerSocketChannel&＃xff0c;异步的服务器端 TCP Socket 连接。
  3 NioDatagramChannel&＃xff0c;异步的 UDP 连接。
  4 NioSctpChannel&＃xff0c;异步的客户端 Sctp 连接。
  5 NioSctpServerChannel&＃xff0c;异步的 Sctp 服务器端连接&＃xff0c;这些通道涵盖了 UDP 和 TCP 网络 IO 以及文件 IO。

• Selector
  Netty 基于 Selector 对象实现 I/O 多路复用&＃xff0c;通过 Selector 一个线程可以监听多个连接的 Channel事件。
  当向一个 Selector 中注册 Channel 后&＃xff0c;Selector 内部的机制就可以自动不断地查询(Select) 这些注册的 Channel 是否有已就绪的 I/O 事件&＃xff08;例如可读&＃xff0c;可写&＃xff0c;网络连接完成等&＃xff09;&＃xff0c;这样程序就可以很简单地使用一个线程高效地管理多个 Channel 。

• NioEventLoop
  NioEventLoop 中维护了一个线程和任务队列&＃xff0c;支持异步提交执行任务&＃xff0c;线程启动时会调用NioEventLoop 的 run 方法&＃xff0c;执行 I/O 任务和非 I/O 任务&＃xff1a;
  I/O 任务&＃xff0c;即 selectionKey 中 ready 的事件&＃xff0c;如 accept、connect、read、write 等&＃xff0c;由processSelectedKeys 方法触发。
  非 IO 任务&＃xff0c;添加到 taskQueue 中的任务&＃xff0c;如 register0、bind0 等任务&＃xff0c;由 runAllTasks 方法触发。

• NioEventLoopGroup
  NioEventLoopGroup&＃xff0c;主要管理 eventLoop 的生命周期&＃xff0c;可以理解为一个线程池&＃xff0c;内部维护了一组线程&＃xff0c;每个线程(NioEventLoop)负责处理多个 Channel 上的事件&＃xff0c;而一个 Channel 只对应于一个线程。

• ChannelHandler
  ChannelHandler 是一个接口&＃xff0c;处理 I/O 事件或拦截 I/O 操作&＃xff0c;并将其转发到其 ChannelPipeline(业
  务处理链)中的下一个处理程序。
  ChannelHandler 本身并没有提供很多方法&＃xff0c;因为这个接口有许多的方法需要实现&＃xff0c;方便使用期间&＃xff0c;
  可以继承它的子类&＃xff1a;
  1 ChannelInboundHandler 用于处理入站 I/O 事件。
  2 ChannelOutboundHandler 用于处理出站 I/O 操作。
  或者使用以下适配器类&＃xff1a;
  1 ChannelInboundHandlerAdapter 用于处理入站 I/O 事件。
  2 ChannelOutboundHandlerAdapter 用于处理出站 I/O 操作。

• ChannelHandlerContext
  保存 Channel 相关的所有上下文信息&＃xff0c;同时关联一个 ChannelHandler 对象。

• ChannelPipline
  保存 ChannelHandler 的 List&＃xff0c;用于处理或拦截 Channel 的入站事件和出站操作。
  ChannelPipeline 实现了一种高级形式的拦截过滤器模式&＃xff0c;使用户可以完全控制事件的处理方式&＃xff0c;以
  及 Channel 中各个的 ChannelHandler 如何相互交互。
  在 Netty 中每个 Channel 都有且仅有一个 ChannelPipeline 与之对应&＃xff0c;它们的组成关系如下&＃xff1a;

  

  一个 Channel 包含了一个 ChannelPipeline&＃xff0c;而 ChannelPipeline 中又维护了一个由
  ChannelHandlerContext 组成的双向链表&＃xff0c;并且每个 ChannelHandlerContext 中又关联着一个
  ChannelHandler。
  read事件(入站事件)和write事件(出站事件)在一个双向链表中&＃xff0c;入站事件会从链表 head 往后传递到最
  后一个入站的 handler&＃xff0c;出站事件会从链表 tail 往前传递到最前一个出站的 handler&＃xff0c;两种类型的
  handler 互不干扰

服务器

public class ChatServer
public static void main(String[] args) throws Exception
EventLoopGroup bossGroup &＃61; new NioEventLoopGroup(1);
EventLoopGroup workerGroup &＃61; new NioEventLoopGroup();
try
ServerBootstrap bootstrap &＃61; new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
.childHandler(new ChannelInitializer()
&＃64;Override
protected void initChannel(SocketChannel ch) throws Exception
ChannelPipeline pipeline &＃61; ch.pipeline();
//向pipeline加入字符串解码器
pipeline.addLast(new StringDecoder());
//向pipeline加入字符串编码器
pipeline.addLast(new StringEncoder());
//加入自己的业务处理handler
pipeline.addLast(new ChatServerHandler());

);
System.out.println("聊天室server启动");
ChannelFuture channelFuture &＃61; bootstrap.bind(9000).sync();
//关闭通道
channelFuture.channel().closeFuture().sync();
finally
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();




服务器处理器

public class ChatServerHandler extends SimpleChannelInboundHandler
//全局通道组&＃xff0c;所有通道都会加入到该组
private static ChannelGroup channelGroup &＃61; new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

//表示 channel 处于就绪状态, 提示上线
&＃64;Override
public void channelActive(ChannelHandlerContext ctx) throws Exception
Channel channel &＃61; ctx.channel();
//将该客户加入聊天的信息推送给其它在线的客户端
channelGroup.writeAndFlush(channel.remoteAddress() &＃43; " 上线了\\n");
//将当前 channel 加入到 channelGroup
channelGroup.add(channel);
System.out.println(ctx.channel().remoteAddress() &＃43; " 上线了\\n");

//表示 channel 处于不活动状态, 提示离线了
&＃64;Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception
//将客户离开信息推送给当前在线的客户
channelGroup.writeAndFlush(ctx.channel().remoteAddress() &＃43; " 下线了\\n");
System.out.println(ctx.channel().remoteAddress() &＃43; " 下线了\\n");

//读取数据
&＃64;Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception
//将客户端的消息发送给所有用户
channelGroup.writeAndFlush(ctx.channel().remoteAddress() &＃43; "说&＃xff1a;" &＃43; msg &＃43; "\\n");

&＃64;Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception
//关闭通道
ctx.close();



客户端

public class ChatClient
public static void main(String[] args) throws Exception
EventLoopGroup group &＃61; new NioEventLoopGroup();
try
Bootstrap bootstrap &＃61; new Bootstrap()
.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer()
&＃64;Override
protected void initChannel(SocketChannel ch) throws Exception
ChannelPipeline pipeline &＃61; ch.pipeline();
pipeline.addLast(new StringEncoder());
pipeline.addLast(new StringDecoder());
//设置客户端处理器
pipeline.addLast(new ChatClientHandler());

);
ChannelFuture channelFuture &＃61; bootstrap.connect("127.0.0.1", 9000).sync();
//得到 channel
Channel channel &＃61; channelFuture.channel();
//客户端输入信息
Scanner scanner &＃61; new Scanner(System.in);
while (scanner.hasNextLine())
String msg &＃61; scanner.nextLine();
//通过 channel 发送到服务器端
channel.writeAndFlush(msg);

finally
group.shutdownGracefully();




客户端处理器

public class ChatClientHandler extends SimpleChannelInboundHandler
//读取服务器发来的数据
&＃64;Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception
System.out.println(msg.trim());